package com.tensai.study.java.newconcept.typereactiveprogram.reactor.sink;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SynchronousSink;

import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiFunction;

/**
 * @author Tensai
 * @see reactor.core.publisher.SynchronousSink#next(Object)
 */
public class NextTest {

	@Test
	public void test() {

		BiFunction<Integer, SynchronousSink<String>, Integer> bf = (state, sink) -> {
			sink.next("3 x " + state + " = " + 3 * state);
			if (state == 10) {
				sink.complete();
			}
			return state + 1;
		};

		BiFunction<Integer, SynchronousSink<String>, Integer> bf1 =  new BiFunction<Integer, SynchronousSink<String>, Integer>() {
			@Override
			public Integer apply(Integer state, SynchronousSink<String> sink) {
					sink.next("3 x " + state + " = " + 3 * state);
					if (state == 10) {
						sink.complete();
					}
					return state + 1;
				}
			};

		Flux<String> flux = Flux.generate(
				() -> 0,
				(state, sink) -> {
					sink.next("3 x " + state + " = " + 3 * state);
					if (state == 10) {
						sink.complete();
					}
					return state + 1;
				});
		flux.subscribe();
	}

	@Test
	public void test1() {
		Flux<String> flux = Flux.generate(
				AtomicLong::new,
				(state, sink) -> {
					System.out.print(state);
					System.out.print("\t");
					long i = state.getAndIncrement();
					System.out.print(i);
					sink.next("3 x " + i + " = " + 3 * i);
					if (i == 10) {
						sink.complete();
					}
					System.out.println();
					return state;
				});
		flux.subscribe();
	}

	@Test
	public void test2() {
		Flux<String> flux = Flux.generate(
				AtomicLong::new,
				(state, sink) -> {
					long i = state.getAndIncrement();
					sink.next("3 x " + i + " = " + 3 * i);
					if (i == 10) {
						sink.complete();
					}
					return state;
				},
				(state) -> System.out.println("state: " + state));
		flux.subscribe();
	}

}
